package com.gzc.just.play.server;

import com.beust.jcommander.JCommander;
import com.gzc.just.play.Utils;
import com.gzc.just.play.proto.War3Proto.*;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Arrays;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * Server类是整个Warcraft III网络连接系统的核心服务器组件，负责协调客户端之间的连接和数据转发。让我详细解释其作用和工作流程：<p>
 * （一）Server类的作用<p>
 * Server类主要承担以下几个功能：<p>
 * 1.客户端连接管理：监听并接受客户端连接，为每个客户端分配唯一的虚拟IP地址<p>
 * 2.IP地址分配：通过IPPool类管理虚拟IP地址池，为每个连接的客户端分配IP地址<p>
 * 3.客户端信息同步：维护客户端列表，当有新客户端加入时通知所有客户端更新列表<p>
 * 4.数据转发中介：作为客户端之间的中转站，处理TCP和UDP数据的转发<p>
 * 5.连接协调：协调魔兽争霸游戏客户端之间的连接建立过程<p>
 *
 * （二）工作流程详解<p>
 * 1. 服务器启动<p>
 * 服务器在8888端口监听客户端TCP连接.<p>
 * 初始化虚拟线程池和IP地址池.<p>
 * 准备接受客户端连接.<p>
 *
 * 2. 客户端连接处理(handleClientConnect)<p>
 * 当客户端连接到服务器时：<p>
 * 从IPPool中分配一个唯一的虚拟IP地址给该客户端.<p>
 * 将该客户端的信息添加到connectionMap中.<p>
 * 向客户端发送当前所有已连接客户端的IP列表.<p>
 * 通知所有已连接客户端更新客户端列表.<p>
 *
 * 3. 魔兽争霸连接处理(handleWar3Connect)<p>
 * 当魔兽争霸客户端尝试连接时：<p>
 * 客户端发送目标IP地址.<p>
 * 服务器查找目标客户端并通知其准备连接.<p>
 * 等待目标客户端提供连接通道.<p>
 * 建立双向数据转发通道.<p>
 *
 * 4. 提供魔兽连接处理(handleProvideWar3)<p>
 * 当客户端提供魔兽争霸连接时：<p>
 * 将连接通道放入目标客户端的队列中.<p>
 * 用于后续的数据转发.<p>
 *
 * 5. 客户端消息处理<p>
 * Client内部的run方法负责：<p>
 * 监听客户端发送的消息.<p>
 * 根据消息类型进行相应处理（如转发给目标客户端）.<p>
 * 当客户端断开连接时清理资源并释放IP地址.<p>
 *
 * 6. 数据转发<p>
 * transferData方法实现两个通道之间的双向数据转发，这是连接不同客户端的关键功能。<p>
 *
 * 整个系统通过这种方式实现了让位于不同网络环境下的玩家能够连接到同一个魔兽争霸游戏中，解决了传统局域网游戏的网络限制问题。
 */
public class Server {
    private static final Logger logger = LogManager.getLogger(Server.class);

    // 使用虚拟线程池处理并发任务
    private static final ExecutorService executor =
            Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("vThread-", 1).factory());
    /**
     * 保存所有已连接的客户端,key:war3Ip,value:client
     */
    public static final ConcurrentHashMap<String, ClientSession> connectionMap = new ConcurrentHashMap<>();
    // IP地址池，用于分配给新连接的客户端
    public static final IPPool ipPool = new IPPool();

    private static volatile String SERVER_IP = "0.0.0.0";
    private static volatile int SERVER_PORT;

    public static void main(String[] args) throws Exception {
        if (!initArgs(args)) {
            return;
        }

        try (var serverChannel = AsynchronousServerSocketChannel.open()) {
            var inetSocketAddress = new InetSocketAddress(SERVER_IP, SERVER_PORT);
            logger.info("tcp server start bind,ip={},port={}", inetSocketAddress.getAddress(), inetSocketAddress.getPort());
            serverChannel.bind(inetSocketAddress);
            logger.info("tcp server bind ok,ip={},port={}", inetSocketAddress.getAddress(), inetSocketAddress.getPort());
            // 主循环，持续接受客户端连接
            while (!Thread.interrupted()) {
                var client = serverChannel.accept().get();
                // 每个新连接交给线程池异步处理
                executor.submit(() -> handleClient(client));
            }
        }
    }

    private static boolean initArgs(String[] args) {
        var commandOptions = new ServerCommandOptions();
        JCommander commander = JCommander.newBuilder()
                .programName("war3proxy-server")
                .addObject(commandOptions)
                .build();
        commander.parse(args);
        if (commandOptions.isHelp()) {
            // 打印帮助信息
            commander.usage();
            return false;
        }

        final String ip = commandOptions.getIp();
        if (null != ip && !ip.isEmpty()) {
            logger.info("commandOptions ip={}", ip);
            SERVER_IP = ip;
        }

        final int port = commandOptions.getPort();
        if (port > 0 && port < 65535) {
            logger.info("commandOptions port={}", port);
            SERVER_PORT = port;
        }
        return true;
    }

    // 处理单个客户端连接
    private static void handleClient(AsynchronousSocketChannel clientChannel) {
        try {
            logger.info("New connection from={}", clientChannel.getRemoteAddress());
            var buffer = ByteBuffer.allocate(1024);
            final int len = clientChannel.read(buffer).get();
            if (len > 0) {
                buffer.flip();
                // 读取消息长度
                final int messageLength = buffer.getInt();
                byte[] messageBytes = new byte[messageLength];
                buffer.get(messageBytes);
                // 解析protobuf消息
                War3Message message = War3Message.parseFrom(messageBytes);
                logger.info("connection from={},read msgType={},msg={}",
                        clientChannel.getRemoteAddress(), message.getType(), message.toString());

                switch (message.getType()) {
                    case CLIENT_CONNECT: {
                        // 客户端连接
                        handleClientConnect(clientChannel);
                        break;
                    }
                    case WAR3_CONNECT: {
                        // 魔兽连接
                        handleWar3Connect(message, clientChannel);
                        break;
                    }
                    case PROVIDE_WAR3: {
                        // 提供魔兽连接
                        handleProvideWar3(message, clientChannel);
                        break;
                    }
                    default: {
                        logger.error("Unknown command={}", message.getType());
                        clientChannel.close();
                        break;
                    }
                }
                buffer.clear();
            } else if (len < 0) {
                clientChannel.close();
            }
        } catch (Exception e) {
            logger.error("handleClient error", e);
            Utils.closeAsynchronousSocketChannel(clientChannel);
        }
    }

    // 处理客户端连接命令，分配IP并注册到connectionMap
    private static void handleClientConnect(AsynchronousSocketChannel clientChannel) throws Exception {
        String war3Ip = ipPool.acquireIP();
        String otherWar3Ip = connectionMap.keySet().stream()
                .map(Object::toString)
                .collect(Collectors.joining("#"));
        logger.info("acquireWar3Ip={},otherWar3Ip={}", war3Ip, otherWar3Ip);

        // 创建war3 IP分配消息
        IPAssignment ipAssignment = IPAssignment.newBuilder()
                .setAssignedIp(war3Ip)
                .addAllOtherIps(Arrays.asList(otherWar3Ip.split("#")))
                .build();
        War3Message message = War3Message.newBuilder()
                .setType(MessageType.CLIENT_CONNECT_RESULT)
                .setPayload(ipAssignment.toByteString())
                .build();
        Utils.sendProtobufMessage(clientChannel, message);
        var client = new ClientSession(war3Ip, clientChannel);
        connectionMap.put(war3Ip, client);
        executor.submit(client);

        for (ClientSession value : connectionMap.values()) {
            // 更新客户端列表
            executor.submit(value::updateClientList);
        }
    }

    // 处理魔兽客户端连接命令，建立数据转发通道
    private static void handleWar3Connect(War3Message message, AsynchronousSocketChannel channel)
            throws Exception {
        String dstIp = message.getTargetIp();
        ClientSession dstClientSession = connectionMap.get(dstIp);
        if (null == dstClientSession) {
            logger.error("not found dstClient from dstIp={}", dstIp);
            channel.close();
            return;
        }

        dstClientSession.notification();
        try {
            // 等待目标客户端提供的Socket通道
            try (var dstChannel = dstClientSession.pollQueue()) {
                if (null == dstChannel) {
                    logger.error("transferData fail,dstChannel is null,dstIp={},dstClient war3Ip={}", dstIp, dstClientSession.getWar3Ip());
                    channel.close();
                    return;
                }
                logger.info("transferData start,dstChannel={},channel={}", dstChannel, channel);
                // 双向数据转发
                Future<?> taskFuture1 = executor.submit(() -> Utils.transferData(dstChannel, channel));
                Future<?> taskFuture2 = executor.submit(() -> Utils.transferData(channel, dstChannel));
                taskFuture1.get();
                taskFuture2.get();
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            channel.close();
        }
    }

    // 处理魔兽服务端提供连接命令，将Socket通道放入队列
    private static void handleProvideWar3(War3Message message, AsynchronousSocketChannel channel) throws Exception {
        String srcIp = message.getSourceIp();
        ClientSession clientSession = connectionMap.get(srcIp);
        if (clientSession != null) {
            logger.info("provideWar3,srcIp={},channel={}", srcIp, channel);
            clientSession.putToQueue(channel);
        } else {
            logger.error("not found client for srcIp={}", srcIp);
            channel.close();
        }
    }
}